package cn.wangjie.spark.convert

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 采用反射的方式将RDD转换为DataFrame和Dataset
 * TODO: RDD[CaseClass]
 */
object SparkRDDInferring {
	
	def main(args: Array[String]): Unit = {
		// 创建SparkSession实例对象，通过建造者模式
		val spark: SparkSession = SparkSession.builder()
			// 设置应用名称
			.appName(this.getClass.getSimpleName.stripSuffix("$"))
			// 设置运行模式
			.master("local[2]")
			.getOrCreate()
		// 导入隐式转换函数
		import spark.implicits._
		
		// 加载文本数据
		val rawRatingsRDD: RDD[String] = spark.sparkContext.textFile("datas/ml-100k/u.data")
		
		// TODO: 将RDD中每行数据解析封装CaseClass样例类中
		val ratingsRDD: RDD[MovieRating] = rawRatingsRDD
			.filter(line => null != line && line.trim.split("\\t").length == 4)
			.mapPartitions{iter =>
				iter.map{line =>
					val arr = line.trim.split("\\t")
					MovieRating(arr(0), arr(1), arr(2).toDouble, arr(3).toLong)
				}
			}
		
		// TODO：将RDD转化为DataFrame，通过反射推断
		val ratingsDF: DataFrame = ratingsRDD.toDF()
		
		ratingsDF.printSchema()
		ratingsDF.show(10, truncate = false)
		
		// 应用结束，关闭资源
		spark.stop()
	}
	
}
